package com.webank.wedatasphere.qualitis.listener;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.webank.wedatasphere.qualitis.handler.CommonExecuteTaskService;
import com.webank.wedatasphere.qualitis.handler.TaskCalculateService;
import com.webank.wedatasphere.qualitis.rule.dao.repository.SqlTaskRepository;
import com.webank.wedatasphere.qualitis.rule.entity.RuleDetailInfoVo;
import com.webank.wedatasphere.qualitis.rule.entity.SqlTaskInfo;
import com.webank.wedatasphere.qualitis.util.LocalTimeUtils;
import org.apache.commons.httpclient.HttpStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.EventListener;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * 事件监听器
 */
@Component
public class TaskExecuteEventListener {

    private Logger logger = LoggerFactory.getLogger(TaskExecuteEventListener.class);

    /**
     * 执行任务服务
     */
    @Autowired
    private CommonExecuteTaskService commonExecuteTaskService;

    @Autowired
    private TaskCalculateService taskCalculateService;


    /**
     * 任务执行记录
     */
    @Autowired
    private SqlTaskRepository sqlTaskRepository;


    @EventListener
    @Async
    public void aa(TaskExecuteEvent event) {
        logger.info("接收到事件:" + event.getClass());
        // 执行任务
        try {
            executeTask(event.getThatObj(), event.getProjectId(), event.getProjectName(), event.getBeginTime(), event.getSortValue());
        } catch (InterruptedException e) {
            logger.error(e.getMessage(), e);
        }
        logger.info("=========================任务执行完成=========================");
    }

    public void executeTask(List<RuleDetailInfoVo> thatObj, Long projectId, String projectName, String beginTime, int sortValue) throws InterruptedException {
        for (RuleDetailInfoVo ruleDetailInfoVo : thatObj) {
            String checkTemplate = ruleDetailInfoVo.getCheckTemplate();
            String sqlPreview = ruleDetailInfoVo.getSqlPreview();
            ResponseEntity<JSONObject> jsonResp = null;
            if (checkTemplate.equals("固定值")) {
                // 底层引擎Spark执行SQL语句|禁止使用线程池异步（底层的Linkis引擎无法批处理）
                jsonResp = commonExecuteTaskService.executeSql(sqlPreview);
            }else {
                String table = sqlPreview.split("where ")[0];
                String where = sqlPreview.split("where ")[1];
                String Time1= LocalTimeUtils.getBeforeDaySimple();
                String Time2= LocalTimeUtils.getCurrentTimeSimple();
                if (checkTemplate.equals("日波动")) {
                    Time1 =  LocalTimeUtils.getBeforeDaySimple();
                }
                if (checkTemplate.equals("月波动")) {
                    Time1 = LocalTimeUtils.getBeforeMonthsSimple();
                }
                if (checkTemplate.equals("周波动")) {
                    Time1 = LocalTimeUtils.getBeforeWeekSimple();
                }
                // 计算当前日期的数据
                String sqlChange = " and create_time > " + Time1 + " ";
                sqlChange += " and create_time <= " +Time2 + " ";
                sqlPreview = table + sqlChange + where;
                // 底层引擎Spark执行SQL语句|禁止使用线程池异步（底层的Linkis引擎无法批处理）
                jsonResp = commonExecuteTaskService.executeSql(sqlPreview);
            }
            if (jsonResp == null) {
                // 底层引擎Spark执行SQL语句|禁止使用线程池异步（底层的Linkis引擎无法批处理）
                jsonResp = commonExecuteTaskService.executeSql(sqlPreview);
            }
            if (jsonResp != null && jsonResp.getStatusCode().value() == HttpStatus.SC_OK) {
                JSONObject body = jsonResp.getBody();
                int status = body.getInteger("status");
                String taskId = body.getJSONObject("data").getInteger("taskID").toString();
                String execID = body.getJSONObject("data").getString("execID");
                if (status == 0) {
                    SqlTaskInfo sqlTaskInfo = new SqlTaskInfo();
                    sqlTaskInfo.setBeginTime(beginTime);
                    sqlTaskInfo.setTaskName(ruleDetailInfoVo.getRuleTemplate() + ":" + ruleDetailInfoVo.getRuleName() + "-执行");
                    sqlTaskInfo.setProjectId(String.valueOf(projectId));
                    sqlTaskInfo.setProjectName(projectName);
                    sqlTaskInfo.setSort(sortValue);
                    sqlTaskInfo.setExecuteTime(LocalTimeUtils.getCurrentTime());
                    sqlTaskInfo.setExecuteStatus("执行中");
                    sqlTaskInfo.setTaskId(taskId);
                    sqlTaskInfo.setExecId(execID);
                    sqlTaskInfo.setSqlPreview(ruleDetailInfoVo.getSqlPreview());
                    sqlTaskInfo.setRuleNewId(ruleDetailInfoVo.getId());
                    sqlTaskInfo.setRuleName(ruleDetailInfoVo.getRuleName());
                    sqlTaskInfo.setCheckFiled(ruleDetailInfoVo.getFieldName());
                    sqlTaskInfo.setCheckRule(ruleDetailInfoVo.getRuleTemplate());
                    sqlTaskInfo.setCheckTable(ruleDetailInfoVo.getTableName());
                    sqlTaskInfo.setCheckStatus("校验中");
                    sqlTaskRepository.save(sqlTaskInfo);
                    // 睡眠等待执行获取状态
                    Thread.sleep(1000);
                    logger.info("进入自旋等待任务完成" + LocalTimeUtils.getCurrentTime());
                    // 循环等待状态执行完成
                    String taskStatus = commonExecuteTaskService.getTaskStatusWaitTime(sqlTaskInfo.getExecId(), System.currentTimeMillis());
                    logger.info("自旋等待结束，获取状态为：" + taskStatus + ";" + LocalTimeUtils.getCurrentTime());
                    switch (taskStatus) {
                        case "Succeed":
                            sqlTaskInfo.setExecuteStatus("执行成功");
                            // 获取执行结果路径
                            JSONObject resultMap = commonExecuteTaskService.getFIleURL(sqlTaskInfo.getTaskId());
                            // 获取文件路径
                            String fileUrl = resultMap.getString("resultLocation");
                            sqlTaskInfo.setExecuteLog(fileUrl);
                            // 获取执行引擎的耗时
                            Long costTime = resultMap.getLong("costTime");
                            sqlTaskInfo.setExecuteTime(costTime.toString());
                            // 获取执行引擎的执行结束时间
                            sqlTaskInfo.setEndTime(LocalTimeUtils.getCurrentTime());
                            // 获取执行引擎的执行结果
                            JSONArray result = commonExecuteTaskService.getResult(fileUrl);
                            JSONArray jsonArray = result.getJSONArray(0);
                            // 获取SQL执行结果
                            String numStrResp = jsonArray.get(0).toString();
                            // 根据规则Id获取规则详情
                            String checkThreshold = ruleDetailInfoVo.getCheckThreshold();
                            String compareType = ruleDetailInfoVo.getCompareType();
                            // 处理固定值的比较结果
                            if (checkTemplate.equals("固定值")) {
                                taskCalculateService.handlerFixValueCompare(compareType, numStrResp, checkThreshold, sqlTaskInfo);
                            }else {
                                // 获取前一天执行的记录数
                                taskCalculateService.handlerDayChangeCompare(compareType, numStrResp, checkThreshold, sqlTaskInfo);
                            }
                            break;
                        case "Failed":
                            sqlTaskInfo.setExecuteStatus("执行失败");
                            // 获取执行结果路径
                            JSONObject resultMapV2 = commonExecuteTaskService.getFIleURL(sqlTaskInfo.getTaskId());
                            // 获取文件路径
                            String fileUrlV2 = resultMapV2.getString("resultLocation");
                            sqlTaskInfo.setExecuteLog(fileUrlV2);
                            // 获取执行引擎的耗时
                            Long costTimeV2 = resultMapV2.getLong("costTime");
                            sqlTaskInfo.setExecuteTime(costTimeV2.toString());
                            // 获取执行引擎的执行结束时间
                            String endTimeV2 = resultMapV2.getString("updatedTime");
                            sqlTaskInfo.setEndTime(endTimeV2);
                            sqlTaskInfo.setResult("校验失败");
                            break;
                        default:
                            logger.error("执行任务状态--进行中");
                            break;
                    }
                    // 更新校验引擎的校验任务状态
                    logger.error("---执行任务完成---进入更新-----");
                    sqlTaskRepository.saveAndFlush(sqlTaskInfo);
                }
            }
        }
    }
}
